{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Exercise 3 - Nested Parallelism\n",
    "\n",
    "**GOAL:** The goal of this exercise is to show how to create nested tasks by calling a remote function inside of another remote function.\n",
    "\n",
    "In this exercise, you will implement the structure of a parallel hyperparameter sweep which trains a number of models in parallel. Each model will be trained using parallel gradient computations.\n",
    "\n",
    "### Concepts for this Exercise - Nested Remote Functions\n",
    "\n",
    "Remote functions can call other functions. For example, consider the following.\n",
    "\n",
    "```python\n",
    "@ray.remote\n",
    "def f():\n",
    "    return 1\n",
    "\n",
    "@ray.remote\n",
    "def g():\n",
    "    # Call f 4 times and return the resulting object IDs.\n",
    "    return [f.remote() for _ in range(4)]\n",
    "\n",
    "@ray.remote\n",
    "def h():\n",
    "    # Call f 4 times, block until those 4 tasks finish,\n",
    "    # retrieve the results, and return the values.\n",
    "    return ray.get([f.remote() for _ in range(4)])\n",
    "```\n",
    "\n",
    "Then calling `g` and `h` produces the following behavior.\n",
    "\n",
    "```python\n",
    ">>> ray.get(g.remote())\n",
    "[ObjectID(b1457ba0911ae84989aae86f89409e953dd9a80e),\n",
    " ObjectID(7c14a1d13a56d8dc01e800761a66f09201104275),\n",
    " ObjectID(99763728ffc1a2c0766a2000ebabded52514e9a6),\n",
    " ObjectID(9c2f372e1933b04b2936bb6f58161285829b9914)]\n",
    "\n",
    ">>> ray.get(h.remote())\n",
    "[1, 1, 1, 1]\n",
    "```\n",
    "\n",
    "**One limitation** is that the definition of `f` must come before the definitions of `g` and `h` because as soon as `g` is defined, it will be pickled and shipped to the workers, and so if `f` hasn't been defined yet, the definition will be incomplete."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from __future__ import absolute_import\n",
    "from __future__ import division\n",
    "from __future__ import print_function\n",
    "\n",
    "import numpy as np\n",
    "import ray\n",
    "import time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ray.init(num_cpus=9, include_webui=False, ignore_reinit_error=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This example represents a hyperparameter sweep in which multiple models are trained in parallel. Each model training task also performs data parallel gradient computations.\n",
    "\n",
    "**EXERCISE:** Turn `compute_gradient` and `train_model` into remote functions so that they can be executed in parallel. Inside of `train_model`, do the calls to `compute_gradient` in parallel and fetch the results using `ray.get`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def compute_gradient(data, current_model):\n",
    "    time.sleep(0.03)\n",
    "    return 1\n",
    "\n",
    "def train_model(hyperparameters):\n",
    "    current_model = 0\n",
    "    # Iteratively improve the current model. This outer loop cannot be parallelized.\n",
    "    for _ in range(10):\n",
    "        # EXERCISE: Parallelize the list comprehension in the line below. After you\n",
    "        # turn \"compute_gradient\" into a remote function, you will need to call it\n",
    "        # with \".remote\". The results must be retrieved with \"ray.get\" before \"sum\"\n",
    "        # is called.\n",
    "        total_gradient = sum([compute_gradient(j, current_model) for j in range(2)])\n",
    "        current_model += total_gradient\n",
    "\n",
    "    return current_model\n",
    "\n",
    "assert hasattr(compute_gradient, 'remote'), 'compute_gradient must be a remote function'\n",
    "assert hasattr(train_model, 'remote'), 'train_model must be a remote function'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**EXERCISE:** The code below runs 3 hyperparameter experiments. Change this to run the experiments in parallel."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Sleep a little to improve the accuracy of the timing measurements below.\n",
    "time.sleep(2.0)\n",
    "start_time = time.time()\n",
    "\n",
    "# Run some hyperparaameter experiments.\n",
    "results = []\n",
    "for hyperparameters in [{'learning_rate': 1e-1, 'batch_size': 100},\n",
    "                        {'learning_rate': 1e-2, 'batch_size': 100},\n",
    "                        {'learning_rate': 1e-3, 'batch_size': 100}]:\n",
    "    results.append(train_model(hyperparameters))\n",
    "\n",
    "# EXERCISE: Once you've turned \"results\" into a list of Ray ObjectIDs\n",
    "# by calling train_model.remote, you will need to turn \"results\" back\n",
    "# into a list of integers, e.g., by doing \"results = ray.get(results)\".\n",
    "\n",
    "end_time = time.time()\n",
    "duration = end_time - start_time\n",
    "\n",
    "assert all([isinstance(x, int) for x in results]), \\\n",
    "    'Looks like \"results\" is {}. You may have forgotten to call ray.get.'.format(results)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**VERIFY:** Run some checks to verify that the changes you made to the code were correct. Some of the checks should fail when you initially run the cells. After completing the exercises, the checks should pass.\n",
    "\n",
    "**NOTE:** This exercise is known to have issues on binder that can be resolved by rerunning the above cell a second time."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "assert results == [20, 20, 20]\n",
    "assert duration < 0.5, ('The experiments ran in {} seconds. This is too '\n",
    "                         'slow.'.format(duration))\n",
    "assert duration > 0.3, ('The experiments ran in {} seconds. This is too '\n",
    "                        'fast.'.format(duration))\n",
    "\n",
    "print('Success! The example took {} seconds.'.format(duration))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**EXERCISE:** Use the UI to view the task timeline and to verify that the four tasks were executed in parallel. You can do this as follows.\n",
    "\n",
    "1. Run the following cell to generate a JSON file containing the profiling data.\n",
    "2. Download the timeline file by right clicking on `timeline03.json` in the navigator to the left and choosing **\"Download\"**.\n",
    "3. Open [chrome://tracing/](chrome://tracing/) in the Chrome web browser, click on the **\"Load\"** button and load the downloaded JSON file.\n",
    "\n",
    "To navigate within the timeline, do the following.\n",
    "- Move around by clicking and dragging.\n",
    "- Zoom in and out by holding **alt** and scrolling.\n",
    "\n",
    "**NOTE:** The timeline visualization will only work in **Chrome**."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ray.timeline(filename=\"timeline03.json\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Application: NLP Pipeline\n",
    "\n",
    "To demonstrate the practical applications of nested remote functions, we create a NLP pipeline that analyzes some interesting Wikipedia pages and we speed up this pipeline using nested remote functions. Though this is only a toy example, this example can easily be scaled up with very little changes! \n",
    "\n",
    "For this NLP pipeline, we first use the function `parse_wikipedia` to parse a Wikipedia page on a given topic. Within `parse_wikipedia`, we first tokenize the page using the `tokenize` function and then feed the tokens to `entity_recognizer`, a named entity recognizer. Not satisfied with just the named entities of each language, we also search for the Wikipedia pages of the first 10 unique named entities, recursively until we hit a recursion depth of 2. Finally, we return a pandas dataframe containing all the named entities we found.\n",
    "\n",
    "As an example, we parse the wikipedia entries for python, java, and c++, the languages that are used in ray!\n",
    "\n",
    "**NOTE:** We use the `en_core_web_sm`, a pre-trained model provided by `spacy`. Make sure you have it downloaded by running the following piece of code within a jupyter notebook cell.\n",
    "\n",
    "```\n",
    "!python -m spacy download en_core_web_sm\n",
    "```\n",
    "After running the above code, you may need to restart the notebook kernel."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import modin.pandas as pd\n",
    "import spacy\n",
    "import wikipedia\n",
    "\n",
    "MAX_LINKS = 2\n",
    "MAX_DEPTH = 2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def tokenize(text):\n",
    "    time.sleep(2)\n",
    "    nlp = spacy.load(\"en_core_web_sm\")\n",
    "    return nlp(text)\n",
    "    \n",
    "def entity_recognizer(tokens, topic):\n",
    "    time.sleep(2)\n",
    "    results = []\n",
    "    for token in tokens.ents:\n",
    "        results.append([topic, token.text, token.lemma_, token.label_])\n",
    "    \n",
    "    return results\n",
    "\n",
    "def recursive_wiki_scraper(topic, depth=0):\n",
    "    try:\n",
    "        wiki_page = wikipedia.page(topic)\n",
    "    except:\n",
    "        return []\n",
    "    wiki_links = wiki_page.links[:MAX_LINKS]\n",
    "\n",
    "    page_tokens = tokenize(wiki_page.content)\n",
    "    topic_result = entity_recognizer(page_tokens, topic)\n",
    "    result = []\n",
    "    \n",
    "    if depth < MAX_DEPTH:\n",
    "        for link in wiki_links:\n",
    "            result.extend(recursive_wiki_scraper(link, depth+1))\n",
    "    \n",
    "    result.extend(topic_result)\n",
    "    return result"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now let's try and get some information on the languages that ray is built on!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "start = time.time()\n",
    "\n",
    "languages = [\"Python\", \"Java programming\", \"C++\"]\n",
    "results = []\n",
    "for lang in languages:\n",
    "    results.extend(recursive_wiki_scraper(lang))\n",
    "    \n",
    "duration = time.time() - start\n",
    "print(\"Constructing the dataframe took {} seconds.\".format(duration))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Exercise:** Speed up the above NLP pipeline using ray and its nested remote functions. To do so, it is recommended that you only make `tokenize` and `entity_recognizer` be remote functions and not the `recursive_wiki_scraper`. Try and understand why this is the case. Below you should find the a sample of the results. Feel free to explore the data to find interesting associations to the programming languages that ray is written in!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = pd.DataFrame(results, columns=[\"topic\", \"text\", \"lemma\", \"label\"])\n",
    "df.sample(10)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.7"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": false,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {
    "height": "calc(100% - 180px)",
    "left": "10px",
    "top": "150px",
    "width": "382.391px"
   },
   "toc_section_display": true,
   "toc_window_display": true
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
